package com.navinfo.opentsp.platform.computing.analysis.service;

import com.navinfo.opentsp.platform.computing.analysis.entity.VehicleStatusAdditionDto;
import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.PointForDisconnectAnalysis;
import com.navinfo.opentsp.platform.computing.analysis.entity.disconnectedVehicle.ProcessedInfo;
import com.navinfo.opentsp.platform.computing.analysis.util.CommonUtil;
import com.navinfo.opentsp.platform.computing.analysis.util.DateUtils;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import static com.lc.core.protocol.common.LCStatusType.StatusType.rotation_VALUE;

/**
 * 异常断连/长停断连，断连原因判断<br/>
 * 从数据源获取数据service
 */
public class DisconnectedVehicleLoadDataService implements Serializable {
    /**
     * 读取几天的0200数据
     */
    private static final int DAYS = 7;
    private static final int SECONDS_OF_DAY = 24 * 60 * 60;
    private static final List<String> COL_NAMES;

    static {
        COL_NAMES = Arrays.asList("terminalid", "gpsdate", "status", "statusaddition", "alarm", "signalstrength");
    }

    private final String pointDbName;
    private final String pointTableName;
    private final String saveDbName;

    public DisconnectedVehicleLoadDataService(String pointDbName, String pointTableName, String saveDbName) {
        this.pointDbName = pointDbName;
        this.pointTableName = pointTableName;
        this.saveDbName = saveDbName;
    }

    private static String[] daysRange(String day) {
        long last = DateUtils.convertYYYYMMDD(day);
        String[] result = new String[DAYS];
        for (int i = 0; i < DAYS; i++) {
            result[i] = DateUtils.format(last - i * SECONDS_OF_DAY, DateUtils.DateFormat.YYYYMMDD);
        }
        return result;
    }

    /**
     * 获取上次处理过的断连信息
     */
    public JavaRDD<ProcessedInfo> loadDisconnectionInfo(SparkSession spark) {
        String infoHql = PropertiesUtil.getProperties("disconnectedVehicle.hive.data.hql.info");
        infoHql = String.format(infoHql, this.saveDbName);
        Dataset<Row> dataSet = spark.sql(infoHql);
        return dataSet.toJavaRDD().map(row -> {
            ProcessedInfo p = new ProcessedInfo();
            p.setTid(row.getAs("tid"));
            p.setConnected(row.getAs("connected"));
            p.setLackOfElectricity(row.getAs("lackOfElectricity"));
            p.setModified(row.getAs("modified"));
            p.setPoorSignal(row.getAs("poorSignal"));
            p.setAbnormalDisconnection(row.getAs("abnormalDisconnection"));
            p.setLongStayDisconnection(row.getAs("longStayDisconnection"));
            p.setAlarm(row.getAs("alarm"));
            return p;
        });
    }

    /**
     * 获取对象通信号集合
     */
    public JavaRDD<Long> loadTerminalIds(SparkSession spark, final String day) {
        String terminalHql = PropertiesUtil.getProperties("disconnectedVehicle.hive.data.hql.terminal");
        terminalHql = String.format(terminalHql, this.saveDbName, this.saveDbName, this.saveDbName, day);
        Dataset<Row> dataSet = spark.sql(terminalHql);
        return dataSet.toJavaRDD().map(f -> f.getAs("tid"));
    }

    /**
     * 读0200位置点数据获取必要信息
     */
    public JavaRDD<PointForDisconnectAnalysis> loadPoint(SparkSession spark, JavaSparkContext jsc, JavaRDD<Long> terminalIds, final String day) {
        if (terminalIds.isEmpty()) {
            // 通信号为空时返回空
            return jsc.emptyRDD();
        }
        StringBuilder sb = new StringBuilder();
        sb.append(String.format("select %s from ", String.join(",", COL_NAMES)));
        if (StringUtils.isNotEmpty(this.pointDbName)) {
            sb.append(this.pointDbName).append(".");
        }
        sb.append(this.pointTableName);
        sb.append(String.format("\n where terminalid in (%s)",
                terminalIds.map(String::valueOf).reduce((v1, v2) -> v1 + "," + v2)));
        sb.append(String.format("\n and part_time in (%s)", String.join(",", daysRange(day))));

        return toJavaRdd(spark.sql(sb.toString()));
    }

    private JavaRDD<PointForDisconnectAnalysis> toJavaRdd(Dataset<Row> dataSet) {
        return dataSet.toJavaRDD().mapPartitions((FlatMapFunction<Iterator<Row>, PointForDisconnectAnalysis>) iterator -> new Iterator<PointForDisconnectAnalysis>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }

                    @Override
                    public PointForDisconnectAnalysis next() {
                        return getPoint(iterator.next());
                    }
                }
        );
    }

    private PointForDisconnectAnalysis getPoint(Row row) {
        PointForDisconnectAnalysis point = new PointForDisconnectAnalysis();
        point.setTid(row.getAs("terminalid"));
        point.setGpsTime(((long) row.getAs("gpsdate")) * 1000L);
        point.setAccStatus((int) CommonUtil.getToBinary(row.getAs("status"), 0));

        // 车辆状态附加信息
        List<VehicleStatusAdditionDto> statusAdditions = VehicleStatusAdditionDto.parseToList(row.getAs("statusaddition"));
        for (VehicleStatusAdditionDto statusAddition : statusAdditions) {
            long statusValue = statusAddition.getStatusValue();
            switch (Integer.parseInt(statusAddition.getTypes())) {
                case rotation_VALUE:
                    // 当前转速
                    point.setRotation(new BigDecimal(statusValue).divide(new BigDecimal(100), 0, BigDecimal.ROUND_HALF_UP).intValue());
                    break;
                default:
                    break;
            }
        }

        // 报警标志位
        long alarm = row.getAs("alarm");
        point.setAlarm(alarm);
        point.setUndervoltage((int) CommonUtil.getToBinary(alarm, 7));
        point.setPowerDown((int) CommonUtil.getToBinary(alarm, 8));
        // 无线通信网络信号强度
        point.setSignalStrength(row.getAs("signalstrength"));

        return point;
    }
}
